package cn.edu.xmu.datamining.tangzk.mralgos.parallelbfs;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class ParalleBFSMapper extends
		Mapper<LongWritable, Text, IntWritable, GraphNodeWritable> {

	protected void map(
			LongWritable key,
			Text value,
			org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, IntWritable, GraphNodeWritable>.Context context)
			throws java.io.IOException, InterruptedException {
		// value format: id \t distance: id1, id2, ...
		String[] flds = value.toString().split("\t");
		if (flds.length < 2) {
			return;
		}
		int nodeId = Integer.valueOf(flds[0]);
		GraphNodeWritable val = new GraphNodeWritable(flds[1]);
		// output the graph structure
		context.write(new IntWritable(nodeId), val);

		int distance = val.getDistance() == Integer.MAX_VALUE ? Integer.MAX_VALUE
				: val.getDistance() + 1;
		for (int id1 : val.getAdjNodes()) {
			// output the distance
			context.write(new IntWritable(id1), new GraphNodeWritable(distance));
		}
	};

}
